package com.atguigu.day05;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

public class Example5 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SourceFunction<String>() {
                    @Override
                    public void run(SourceContext<String> sourceContext) {
                        // 发送的数据是字符串`“hello”`，数据的事件时间是`1000L`
                        // Watermark(-MAX) run方法开始执行之前发送的
                        // "hello", 1000L
                        // Watermark(10002 * 1000L)
                        // Watermark(+MAX) run方法执行完毕发送的
                        sourceContext.collectWithTimestamp("hello", 1000L);
                        sourceContext.emitWatermark(new Watermark(10002 * 1000L));
                    }

                    @Override
                    public void cancel() {

                    }
                })
                .keyBy(r -> r)
                .process(new KeyedProcessFunction<String, String, String>() {
                    @Override
                    public void processElement(String in, Context context, Collector<String> collector) {
                        // 输入数据的事件时间戳
                        long eventTime = context.timestamp();
                        context.timerService().registerEventTimeTimer(eventTime + 10000 * 1000L);
                        context.timerService().registerEventTimeTimer(eventTime + 10001 * 1000L);
                        context.timerService().registerEventTimeTimer(eventTime + 10002 * 1000L);
                        context.timerService().registerEventTimeTimer(eventTime + 10003 * 1000L);

                        collector.collect("输入数据：" + in + ", 事件时间戳是：" + eventTime + ", 注册的定时器的时间戳是：" +
                                "" + (eventTime + 10000 * 1000L) + "，" +
                                "" + (eventTime + 10001 * 1000L) + ", " +
                                "" + (eventTime + 10002 * 1000L) + ", " +
                                "" + (eventTime + 10003 * 1000L) + ", " +
                                "当前process算子的并行子任务的水位线是：" +
                                "" + context.timerService().currentWatermark());
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
                        out.collect("定时器的时间戳是：" + timestamp + ", 当前process算子并行子任务的水位线是：" +
                                "" + ctx.timerService().currentWatermark());
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
